// ThreadPool.cpp: implementation of the CThreadPool class.
//
//////////////////////////////////////////////////////////////////////

#include "ThreadPool.h"

namespace Model
{

	//////////////////////////////////////////////////////////////////////
	// Construction/Destruction
	//////////////////////////////////////////////////////////////////////

	CThreadPool::CThreadPool()
	{
		m_iDataCount = 0 ;
		m_threadCount = 0;
	}

	CThreadPool::~CThreadPool()
	{
		pthread_cond_destroy(&m_condition);
		pthread_mutex_destroy(&m_mutex);
	}
	
	int CThreadPool::CreatePool( unsigned int threadCount, int timer , void * param )
	{
		if( 0 != InitThreads(timer, param) )
		{
			return -1;
		}

		pthread_t tid;

		for(unsigned int i = 0 ; i< threadCount; i++)
		{
			if( 0 != pthread_create(&tid,NULL,ThreadProc, &m_struThreadParam ))
			{
				StopPool();
				m_mapTids.clear();
				return -1;
			}
			m_mapTids.insert(pair<int,pthread_t> (i, tid ));
		}

		sleep(1);

		m_threadCount = threadCount;
		return 0 ;
	}

	void * CThreadPool::ThreadProc( void * param )
	{
		thread_param_struct *pParamStru = (thread_param_struct*) param;

		map<int , pthread_t >* pTids = pParamStru->pTids;
		map<int , pthread_t >::iterator iter;

		//当前线程序号
		int index = -1;

		//当前线程状态及处理数据的参数副本
		struData dataParam;

		dataParam.pBusyNum = NULL;
		dataParam.bDataRemain = NULL;
		dataParam.pCond = pParamStru->pCond;
		dataParam.pLocker = pParamStru->pMutex;

		pthread_t tid = pthread_self();

		for(iter =pTids->begin();iter!=pTids->end();iter++)
		{
			if(tid==iter->second)
			{
				index = iter->first;
				break;
			}
		}
		if(index==-1)
		{
			return NULL;
		}

		while( !pParamStru->bExit )
		{
			//设置了定时唤醒
			if( pParamStru->timer > 0 ) 
			{
				pParamStru->pThis->mySleep( pParamStru->timer );
				if(pParamStru->pDataList->empty())
					continue;
			}
			pthread_mutex_lock(pParamStru->pMutex);
			pParamStru->iBusy--;
			if(dataParam.pBusyNum)
			{
				(*dataParam.pBusyNum)--;
				if((*dataParam.pBusyNum)==0&&!(*dataParam.bDataRemain) )
				{
					pthread_mutex_lock(dataParam.pLocker);
					*(dataParam.bReady) = true;
					//处理完毕，发信号
					pthread_cond_signal(dataParam.pCond);
					pthread_mutex_unlock(dataParam.pLocker);
					if(pParamStru->pDataList->empty())
					{
						list<struData>().swap(*pParamStru->pDataList);
					}
				}
			}

			//当待处理数据为空时才开始等待
			while( pParamStru->pDataList->empty() && pParamStru->timer <= 0 )
			{
				pthread_cond_wait(pParamStru->pCond,pParamStru->pMutex); 
			}
			if(pParamStru->bExit)
			{
				pthread_mutex_unlock(pParamStru->pMutex);
				break;
			}
			pParamStru->iBusy++;
			//取出数据
			dataParam = pParamStru->pDataList->front();
			//从待处理队列中移除
			pParamStru->pDataList->pop_front();

			//printf("list_buf size:%d,record size:%d.\n",pParamStru->pDataList->size(),*(pParamStru->pListSize));

			pthread_mutex_unlock(pParamStru->pMutex);

			//处理数据
			if( dataParam.pTheadProc != NULL )
			{
				dataParam.pTheadProc->Process(index,dataParam.param,dataParam.dataSize );
			}

		}
		return NULL;
	}

	void CThreadPool::SetSleepTime( int t )
	{

		pthread_mutex_lock(&m_mutex);

		m_struThreadParam.timer = t ;

		pthread_mutex_unlock(&m_mutex);

		return ;
	}
	void CThreadPool::mySleep(int ms)
	{
		struct timeval delay;
		delay.tv_sec = 0;
		delay.tv_usec = ms * 1000; // 1 ms
		select(0, NULL, NULL, NULL, &delay);
	}

	int CThreadPool::Dispatch( struData* pData) const
	{
		//传入空指针会检查是否退出
		if(pData==NULL)
		{
			//压入处理队列
			pthread_mutex_lock(&m_mutex);
			struData tmpData;
			tmpData.pTheadProc = NULL;

			m_dataList.push_back(tmpData);
			if(tmpData.pBusyNum)
			{
				(*tmpData.pBusyNum)++;
			}
			//printf("dispatch , list size:%d \n" , m_iDataCount);
			pthread_mutex_unlock(&m_mutex);
		}
		else
		{
			//压入处理队列
			pthread_mutex_lock(&m_mutex);
			m_dataList.push_back( *pData );

			if(pData->pBusyNum)
			{
				(*pData->pBusyNum)++;
			}
			//printf("dispatch , list size:%d \n" , m_iDataCount);
			pthread_mutex_unlock(&m_mutex);

		}

		pthread_cond_signal(&m_condition);

		return 0 ;
	}

	int CThreadPool::DispatchAll( struData* pData )
	{
		//传入空指针会检查是否退出
		if(pData==NULL)
		{
			//压入处理队列
			pthread_mutex_lock(&m_mutex);
			struData tmpData;
			tmpData.pTheadProc = NULL;

			m_dataList.push_back(tmpData);
			//printf("dispatch , list size:%d \n" , m_iDataCount);
			pthread_mutex_unlock(&m_mutex);
		}
		else
		{
			pthread_mutex_lock(&m_mutex);
			m_dataList.push_back( *pData );
			//printf("dispatch all, list size:%d \n" , m_iDataCount);
			pthread_mutex_unlock(&m_mutex);
		}
		pthread_cond_broadcast(&m_condition);	

		return 0 ;
	}

	int CThreadPool::InitThreads(int timer , void* param )
	{
		int error ;
		error = pthread_mutex_init(&m_mutex,NULL);
		if( error != 0 )
		{
			return -1;
		}

		error = pthread_cond_init(&m_condition,NULL);
		if( error != 0 )
		{
			return -1;
		}

		//定时唤醒时间
		m_struThreadParam.timer = timer;
		//传递参数
		m_struThreadParam.param = param;
		//是否接受到退出信号
		m_struThreadParam.bExit = false;
		//tid表
		m_struThreadParam.pTids = &m_mapTids;
		//线程锁
		m_struThreadParam.pMutex = &m_mutex;
		m_struThreadParam.pCond = &m_condition;
		//待处理数据列表
		m_struThreadParam.pDataList = &m_dataList;
		m_struThreadParam.pThis = this;
		return 0;
	}

	bool CThreadPool::isBusy()
	{
		bool bBusy;
		pthread_mutex_lock(&m_mutex);
		if( m_struThreadParam.iBusy>0)
		{
			bBusy = true;
		}
		else
		{
			bBusy = false;
		}
		pthread_mutex_unlock(&m_mutex);

		return bBusy;
	}

	int  CThreadPool::getThreadCount()
	{
		return m_threadCount;
	}

	int CThreadPool::StopPool( void )
	{
		m_struThreadParam.bExit = true;
		SetSleepTime(1);
		for ( unsigned int i = 0 ; i< m_mapTids.size(); i++ )
		{
			DispatchAll();
		}

		int *pExitcode;

		for ( unsigned int i = 0 ; i< m_mapTids.size(); i++ )
		{
			pthread_join( m_mapTids[i] , (void **)&pExitcode);
		}

		m_mapTids.clear();

		while(!m_dataList.empty())
		{
			m_dataList.pop_front();
		}

		return 0 ;
	}

	char * CThreadPool::AllocBuf( int size )
	{
		if( size <= 0 )
		{
			return NULL;
		}
		else
		{
			return new char[size];
		}
	}

	void CThreadPool::FreeBuf( char* &data )
	{
		if( data != NULL )
		{
			delete [] data ;
			data  = NULL;
		}
	}
}
